package com.cctc.rds.raft.core.rpc.election.impl;

import com.cctc.rds.raft.core.RaftCluster;
import com.cctc.rds.raft.core.log.Log;
import com.cctc.rds.raft.core.log.LogService;
import com.cctc.rds.raft.core.node.Node;
import com.cctc.rds.raft.core.node.NodeRole;
import com.cctc.rds.raft.core.rpc.RpcClientRouter;
import com.cctc.rds.raft.core.rpc.election.*;
import com.cctc.rds.raft.core.util.CountDownWaiter;
import com.cctc.rds.raft.core.util.IdUtils;

import java.io.Closeable;
import java.io.IOException;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Raft 竞选 [拉票] 实现类。
 * 前提：当 Raft 节点空闲时间大于等于 electionMs 选举周期时，将触发自身的竞选 [拉票] 逻辑。
 * 1. 开启竞选，重置拉票计数器：
 * 1）将自身节点角色设置为候选人；
 * 2）置新当前任期号；
 * 3）根据新任期号获取投票器，并将第一票投给自己。
 * 2. 向其他所有节点发送拉票请求，获取选票；
 * 3. 获取其他所有节点的投票结果：
 * --> 如果得票超过半数，竞选成功，更新自身角色。
 * 1）将自身角色设置为领导人；
 * 2）更新自身当前任期号为全集群最新的任期号；
 * 3）将其他节点设置为跟随者。
 * --> 如果得票少于半数，竞选失败，集群中一定出现了新的领导者，更新自身角色。
 * 1）将自身角色设置为跟随者；
 * 2）更新自身当前任期号，与领导人的任期号（全集群最新的任期号）保持一致。
 */
public class ElectionFunctionImpl implements ElectionRpcScheduler.ElectionFunction, Closeable {
    private int nodeId;
    private LogService logService;
    private RpcClientRouter<ElectionRpc> router;
    private ElectionListener electionListener;
    private ExecutorService electionWorkerGroup;

    public ElectionFunctionImpl(int nodeId, LogService logService, RpcClientRouter<ElectionRpc> router, ElectionListener electionListener) {
        this.nodeId = nodeId;
        this.logService = logService;
        this.router = router;
        this.electionListener = electionListener;
        this.electionWorkerGroup = Executors.newFixedThreadPool(
                RaftCluster.nodeCount() - 1,
                r -> new Thread(r, "elect-rpc-worker-" + IdUtils.newId()));
    }

    @Override
    public void startElection() {
        Node curNode = RaftCluster.getNode(nodeId);
        electionListener.startElection();
        if (!curNode.resetVoteNumber()) return;
        while (!electionWorkerGroup.isShutdown() && !electionWorkerGroup.isTerminated()) {
            if (curNode.getRole() != NodeRole.Candidate) break;
            ElectionRpcRequest request = ElectionRpcRequest.builder()
                    .nodeId(curNode.getId())
                    .term(curNode.getCurTerm())
                    .build();
            Log log = logService.index(curNode.curCommitLogIndex());
            if (log == null) {
                request.setLastCommitLogIndex(-1);
                request.setLastCommitLogTerm(-1);
            } else {
                request.setLastCommitLogIndex(log.getIndex());
                request.setLastCommitLogTerm(log.getTerm());
            }
            Set<Node> nodes = RaftCluster.getAllNodes();
            int count = RaftCluster.nodeCount();
            final CountDownWaiter waiter = new CountDownWaiter(count - 1);
            final AtomicLong newLeaderTerm = new AtomicLong(0);
            for (Node toNode : nodes) {
                if (toNode.getId().equals(curNode.getId())) continue;
                electionWorkerGroup.execute(() -> {
                    try {
                        ElectionRpcResponse response = router.route(toNode.getIpPort()).elect(request);
                        if (response.isGranted()) {
                            curNode.incVoteNumber();
                            waiter.countDownSuccess();
                        } else if (response.getTerm() > request.getLastCommitLogIndex()) {
                            newLeaderTerm.set(response.getTerm());
                        }
                    } catch (Exception e) {
                        waiter.countDownException();
                    } finally {
                        waiter.countDown();
                    }
                });
            }
            int half = (count / 2) + 1;
            waiter.await(half - 1);
            if (waiter.getCountException() > half - 1) {
                waiter.await();
                continue;
            }
            if (curNode.getRole() == NodeRole.Candidate) {
                if (curNode.getVoteNumber() >= half) {
                    RaftCluster.getNode(curNode.getId()).asLeader(request.getLastCommitLogTerm());
                    electionListener.endElection(NodeRole.Leader);
                } else if (newLeaderTerm.get() > 0) {
                    RaftCluster.getNode(curNode.getId()).asFollower(newLeaderTerm.get());
                    electionListener.endElection(NodeRole.Follower);
                }
            }
        }
    }

    @Override
    public void close() {
        electionWorkerGroup.shutdown();
    }
}
